package hanlen.mq.starter.core.impl;

import hanlen.mq.starter.bean.MessageWrap;
import hanlen.mq.starter.constants.QueueConstants;
import hanlen.mq.starter.core.RedissonService;
import hanlen.mq.starter.support.IdUtil;
import java.util.HashMap;
import java.util.Map;
import java.util.Optional;
import java.util.concurrent.TimeUnit;
import java.util.function.Function;
import javax.annotation.Resource;
import lombok.extern.slf4j.Slf4j;
import org.redisson.api.RAtomicLong;
import org.redisson.api.RBlockingQueue;
import org.redisson.api.RBucket;
import org.redisson.api.RMapCache;
import org.redisson.api.RTopic;
import org.redisson.api.RedissonClient;
import org.redisson.client.codec.Codec;
import org.redisson.codec.FstCodec;
import org.slf4j.MDC;
import org.springframework.beans.factory.annotation.Value;

/**
 * mq-spring-boot-starter
 *
 * @author MrYe
 * @Description
 * @date 2021-10-26 11:32
 */
@Slf4j
public class DefaultRedissonServiceImpl implements RedissonService {

    @Value("${spring.profiles.active:dev}")
    private String env;

    @Value("${app.id:defaultApp}")
    private String appId;

    @Resource
    private RedissonClient redissonClient;

    private String renameKey(String key) {
        return env + QueueConstants.SEPARATOR + appId + QueueConstants.SEPARATOR + key;
    }

    @Override
    public <T> void addQueue(String name, T data) {
        addDelayedQueue(name, new HashMap<>(), data, 1, TimeUnit.MILLISECONDS);
    }

    @Override
    public <T> void addQueue(String name, Map<String, String> header, T data) {
        addDelayedQueue(name, header, data, 1, TimeUnit.MILLISECONDS);
    }

    @Override
    public <T> void addDelayedQueue(String name, Map<String, String> header, T data, long delay, TimeUnit timeUnit) {
        if (data instanceof MessageWrap) {
            throw new IllegalArgumentException("data type MessageWrap is error");
        }
        final String messageId = Optional.ofNullable(MDC.get(QueueConstants.MESSAGE_ID)).orElse(IdUtil.fastSimpleUUID());
        final MessageWrap message = MessageWrap.builder().messageId(messageId).header(header).body(data).build();

        final String key = renameKey(name);
        redissonClient.getDelayedQueue(redissonClient.getBlockingQueue(key)).offer(message, delay, timeUnit);
        redissonClient.getAtomicLong(QueueConstants.getSizeKey(key)).incrementAndGet();
    }

    @Override
    public <T> void deleteDelayedQueue(String name) {
        final String key = renameKey(name);
        redissonClient.getDelayedQueue(redissonClient.getBlockingQueue(key)).delete();
    }

    @Override
    public <T> RBlockingQueue<T> getBlockingQueue(String name) {
        final String key = renameKey(name);
        return redissonClient.getBlockingQueue(key);
    }

    @Override
    public <V> boolean set(String key, V val) {
        final String cacheKey = renameKey(key);
        try {
            log.debug("Redis set key:{}, forever", cacheKey);
            final RBucket<V> bucket = redissonClient.getBucket(cacheKey);
            bucket.set(val);
            return true;
        } catch (Exception e) {
            log.error("Redis exp key:{} forever {}", cacheKey, e);
            return false;
        }
    }

    @Override
    public <V> boolean set(String key, V val, long expired, TimeUnit timeUnit) {
        final String cacheKey = renameKey(key);
        try {
            log.info("Redis set key:{}, expired:{}, timeUnit:{}", cacheKey, expired, timeUnit);
            final RBucket<V> bucket = redissonClient.getBucket(cacheKey);
            bucket.set(val, expired, timeUnit);
            return true;
        } catch (Exception e) {
            log.error("Redis exp key:{} expired:{} {}", cacheKey, expired, e);
            return false;
        }
    }

    @Override
    public boolean delete(String key) {
        final String cacheKey = renameKey(key);
        try {
            log.debug("Redis delete key:{}", cacheKey);
            redissonClient.getBucket(cacheKey).delete();
            return true;
        } catch (Exception e) {
            log.error("Redis exp delete key:{} {}", cacheKey, e);
            return false;
        }
    }

    @Override
    public <R> R get(String key) {
        final String cacheKey = renameKey(key);
        try {
            log.debug("Redis get key:{}", cacheKey);
            final RBucket<R> bucket = redissonClient.getBucket(cacheKey);
            return bucket.get();
        } catch (Exception e) {
            log.error("Redis exp get key:{} {}", cacheKey, e);
            return null;
        }
    }

    @Override
    public <R> R get(String key, Function<String, R> function, long timeToLive, TimeUnit timeUnit) {
        final String cacheKey = renameKey(key);
        try {
            log.debug("Redis get key:{}", cacheKey);
            final RBucket<R> bucket = redissonClient.getBucket(cacheKey);
            final R r = bucket.get();
            if (r == null) {
                redissonClient.getBucket(cacheKey).set(function.apply(key), timeToLive, timeUnit);
            }
            return r;
        } catch (Exception e) {
            log.error("Redis exp get key:{} {}", cacheKey, e);
            return null;
        }
    }

    @Override
    public RAtomicLong getAtomicLong(String key) {
        final RAtomicLong atomicLong = redissonClient.getAtomicLong(renameKey(key));
        atomicLong.expire(360, TimeUnit.DAYS);
        return atomicLong;
    }

    @Override
    public RMapCache<String, Object> getMapCache(String name) {
        return redissonClient.getMapCache(name);
    }

    @Override
    public void publish(String topicName, Object value) {
        try {
            redissonClient.getTopic(renameKey(topicName), new FstCodec()).publish(value);
        } catch (Exception e) {
            log.error("redis publish error", e);
        }
    }

    @Override
    public void publish(String topicName, Object value, Codec codec) {
        redissonClient.getTopic(renameKey(topicName), codec).publish(value);
    }

    @Override
    public RTopic getTopic(String topicName, Codec codec) {
        return redissonClient.getTopic(renameKey(topicName), codec);
    }

}
